package com.uestc.coprocessor.userLog;

import java.io.IOException;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import com.uestc.coprocessor.userLog.UserLogAggregationProtocol.CountRequest;
import com.uestc.coprocessor.userLog.UserLogAggregationProtocol.CountResponse;

/**
 * 用户日志的地理位置统计
 * 
 * @author lijiong
 *
 */
public class UserLogAggregation extends UserLogAggregationProtocol.ActiveUserLocationCountService
		implements Coprocessor, CoprocessorService {

	private RegionCoprocessorEnvironment env;

	/**
	 * 获取各省各项操作的次数
	 */
	@Override
	public void getCounts(RpcController controller, CountRequest request, RpcCallback<CountResponse> done) {
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("province"));
		String beginDate = request.getBeginDate();
		String endDate = request.getEndDate();
		ArrayList<String> dateList = getDatesBetweenTowDates(beginDate, endDate);
		ArrayList<Filter> filters = new ArrayList<>();
		for (String day : dateList) {
			RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(day));
			filters.add(filter);
		}
		Filter filter = new FilterList(FilterList.Operator.MUST_PASS_ONE, filters);
		scan.setFilter(filter);
		UserLogAggregationProtocol.CountResponse response = null;
		InternalScanner scanner = null;
		LinkedHashMap<String, Long> map = (LinkedHashMap<String, Long>) ProvinceInfo.getMap();

		try {
			scanner = env.getRegion().getScanner(scan);
			List<Cell> results = new ArrayList<Cell>();
			boolean hasMore = false;
			do {
				hasMore = scanner.next(results);
				for (Cell kv : results) {
					String province = Bytes.toString(CellUtil.cloneValue(kv));
					if (province == null || province.isEmpty())
						continue;
					Long value = map.get(province);
					if (value == null) { // 不在中国境内，暂时不统计
						continue;
					} else {
						map.put(province, ++value);
					}
				}
				results.clear();
			} while (hasMore);

			// 生成返回给调用端的字符串
			StringBuilder sb = new StringBuilder();
			// 遍历map
			for (Map.Entry<String, Long> item : map.entrySet()) {
				sb.append(item.getKey());
				sb.append(":");
				sb.append(item.getValue());
				sb.append(";");
			}
			response = UserLogAggregationProtocol.CountResponse.newBuilder().setResult(sb.toString()).build();
		} catch (IOException e) {
			e.printStackTrace();
			ResponseConverter.setControllerException(controller, e);
		} finally {
			if (scanner != null) {
				try {
					scanner.close();
				} catch (IOException ignored) {
				}
			}
		}
		done.run(response);
	}

	@Override
	public Service getService() {
		return this;
	}

	@Override
	public void start(CoprocessorEnvironment arg0) throws IOException {
		if (arg0 instanceof RegionCoprocessorEnvironment) {
			this.env = (RegionCoprocessorEnvironment) arg0;
		} else {
			throw new CoprocessorException("Must be loaded on a table region!");
		}
	}

	@Override
	public void stop(CoprocessorEnvironment arg0) throws IOException {

	}

	/**
	 * 返回两个日期间的日期，包括自身的两天
	 * 
	 * @param time1
	 *            开始时间
	 * @param time2
	 *            结束时间
	 * @return
	 */
	public static ArrayList<String> getDatesBetweenTowDates(String time1, String time2) {
		if (time1 == null || time2 == null || time1.isEmpty() || time2.isEmpty()) {
			time1 = LocalDate.now().toString();
			time2 = time1;
		}
		LocalDate t1 = LocalDate.parse(time1);
		LocalDate t2 = LocalDate.parse(time2);
		ArrayList<String> list = new ArrayList<>();
		while (t1.compareTo(t2) < 0) {
			list.add(t1.toString());
			t1 = t1.plusDays(1);
		}
		list.add(t2.toString());
		return list;
	}

}
